package com.peng.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.ExecutionException;

/**
 * KafkaProducer
 *
 * @author: lupeng6
 * @create: 2023/3/18 13:54
 */
public class MyKafkaTimeConsumer {

    // topic名字
    private static final String TOPIC_NAME = "sample2";
    // 消费者组名字
    private static final String CONSUMER_GROUP = "testGroup1";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        // broker列表
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.14:9092,192.168.1.14:9093,192.168.1.14:9094");
        // consumerGroup
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP);
        // key反序列化方式
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // value反序列化方式
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // consumer 一次拉取多少条数据，默认500，如果consumer的等待时间内没有拿到500则继续poll
        // 如果拿到了500条数据则执行消费者的逻辑
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        // 如果两次poll的时间大于30秒则kafka会任务消费者的消费能力差，将consumer踢出消费者组，将partition给其他消费者
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
        // consumer给broker发送心跳的时间间隔
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        // kafka如果超过10秒没有收到消费者的心跳，则会吧消费者提出组，进行rebalance，将分区交给其他的消费者
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            // 获取分区信息，然后遍历分区设置每个分区开始消费时间，kafka会根据时间查询offset。
            // 获取到partition 和 offset之后指定consumer的消费partition和offset
            List<PartitionInfo> partitionInfos = consumer.partitionsFor(TOPIC_NAME, Duration.ofMillis(5000));
            final long beginTime = new Date().getTime() - 60 * 60 * 1000;
            Map<TopicPartition, Long> partition2BeginTime = new HashMap<>();
            for (PartitionInfo p : partitionInfos) {
                partition2BeginTime.put(new TopicPartition(TOPIC_NAME, p.partition()), beginTime);
            }
            Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes = consumer.offsetsForTimes(partition2BeginTime);
            for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsetsForTimes.entrySet()) {
                TopicPartition partition = entry.getKey();
                OffsetAndTimestamp timeOffset = entry.getValue();
                if (partition == null || timeOffset == null) {
                    continue;
                }
                long offset = timeOffset.offset();
                System.out.printf("时间偏移 partition:%s offset:%s%n", partition.partition(), offset);
                consumer.assign(Arrays.asList(partition));
                consumer.seek(partition, offset);
            }

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("收到消息：partition=%s offset=%s key=%s value=%s%n",
                            record.partition(), record.offset(), record.key(), record.value());
                }
                if (records.count() > 0) {
                    // 手动异步提交offset，当前线程提交offset不会阻塞
                    consumer.commitAsync(new OffsetCommitCallback() {
                        @Override
                        public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                            if (exception != null) {
                                System.err.println("Commit failed for " + offsets);
                                System.err.println("Commit failed exception: " + Arrays.toString(exception.getStackTrace()));
                            }
                        }
                    });
                }
            }
        }
    }
}
